package com.xxmw.transaction.service.app.component;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxmw.transaction.common.utils.DeflateUtil;
import com.xxmw.transaction.enumeration.BourseEnum;
import com.xxmw.transaction.model.app.BoursePrice;
import com.xxmw.transaction.model.app.Symbol;
import com.xxmw.transaction.service.app.SymbolService;
import com.xxmw.transaction.service.support.AbstractWebSocketAdapter;
import com.xxmw.transaction.service.support.SymbolSubWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public final class OkexTickerApplication extends AbstractWebSocketAdapter implements SymbolSubWebSocket {

	private static final String OX_URL = "wss://real.okex.com:8443/ws/v3";

	private static final String OP = "op";

	private static final String PARAMS = "args";

	private static final String SUBSCRIBE = "subscribe";

	private static final String UNSUBSCRIBE = "unsubscribe";

	private static final String SUBSCRIBE_VALUE = "spot/ticker";

	private long period = 3000;

	@Autowired
	private SymbolService symbolService;

	@Override
	public void afterPropertiesSet() throws Exception {
		super.afterPropertiesSet(new OkexWebSocketClient(new URI(OX_URL), new Draft_6455(), null, 5000));
		scheduled.scheduleWithFixedDelay(() -> heartBeat(), period, period, TimeUnit.MILLISECONDS);
	}

	@Override
	public void destroy() throws Exception {
		this.client.close();
		scheduled.shutdown();
	}

	@Override
	public void subscribe(Symbol symbol) {
		//发送订阅消息
		Set<String> subSet = new HashSet<String>();
		subSet.add(SUBSCRIBE_VALUE+":"+BourseEnum.OKEX.getSymbolFormat().symbolFormat(symbol));
		sendMessage(subSet, SUBSCRIBE);
	}

	@Override
	public void unSubscribe(Symbol symbol) {
		//发送取消订阅消息
		Set<String> subSet = new HashSet<String>();
		subSet.add(SUBSCRIBE_VALUE+":"+BourseEnum.OKEX.getSymbolFormat().symbolFormat(symbol));
		sendMessage(subSet, UNSUBSCRIBE);
	}

	@Override
	public void subscribeList(Set<Symbol> subSet) {
		subSet.forEach(item -> subscribe(item));
	}

	@Override
	public BoursePrice getSymbolPrice (String symbol) {
		return symbol_map.get(symbol);
	}

	@Override
	public void cleanSymbolPrice(String symbol) {
		symbol_map.remove(symbol);
	}

	private void sendMessage (Set<String> subSet, String method) {
		JSONObject json = new JSONObject();
		json.put(OP, method);
		json.put(PARAMS, subSet);
		this.client.send(json.toJSONString());
	}


	private class OkexWebSocketClient extends WebSocketClient {

		public OkexWebSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders,
								   int connectTimeout) {
			super(serverUri, protocolDraft, httpHeaders, connectTimeout);
		}

		@Override
		public void onOpen(ServerHandshake handshakedata) {
			log.info("OKEX WebSocket接口: {} 链接成功.", this.getURI());
			Set<Symbol> subSet = symbolService.listByBourse(BourseEnum.OKEX.getValue());
			//Set<Symbol> subSet = null;//symbolSubService.listByBourse(3);
			if (subSet != null && !subSet.isEmpty()) {
				//Set<String> topicSet = subSet.stream().map(item -> SUBSCRIBE_VALUE+":"+ BourseEnum.OKEX.getSymbolFormat().symbolFormat(item)).collect(Collectors.toSet());
				log.info("OK订阅开始..{}",subSet);
				subscribeList(subSet);
			}
		}

		@Override
		public void onMessage(String s) {
			log.info("接受到ok的ws数据：{}，未作处理", s);
		}

		@Override
		public void onMessage(ByteBuffer byteBuffer) {
			byte[] temp = byteBuffer.array();
			String res = DeflateUtil.decompress(temp);
			// log.info("接收到OKEX websocket消息：{}", res);
			if (res.contains("best_bid")) {
				JSONObject resObject = JSONObject.parseObject(res);
				JSONArray array = resObject.getJSONArray("data");
				if (array != null) {
					JSONObject data = array.getJSONObject(0);
					String symbol = data.getString("instrument_id");
					symbol = BourseEnum.OKEX.getSymbolFormat().symbolStandardization(symbol);
					data.put("updateTime",new Date());
//					log.info("OK获取{}买一价为：{}，卖一价为：{}", symbol, data.getString("best_bid"), data.getString("best_ask"));
					BoursePrice price = BoursePrice.builder().buyPrice(data.getBigDecimal("best_bid")).buyNum(data.getBigDecimal("best_bid_size")).sellPrice(data.getBigDecimal("best_ask")).sellNum(data.getBigDecimal("best_ask_size")).data(data.toJSONString()).updateTime(new Date()).build();
					// log.info("OK获取到【***{}***】的实时价格为：{}", symbol, price.toString());
					symbol_map.put(symbol, price);
				}
			} else {
				log.info("接受到ok的ws数据：{}，未作处理", res);
			}
		}


		@Override
		public void onClose(int i, String s, boolean b) {
			log.info("已关闭");
		}

		@Override
		public void onError(Exception e) {
			log.info("OK链接错误，{}", e);
		}

//		private String uncompress(byte[] bytes) {
//			try (final ByteArrayOutputStream out = new ByteArrayOutputStream();
//				 final ByteArrayInputStream in = new ByteArrayInputStream(bytes);
//				 final Deflate64CompressorInputStream zin = new Deflate64CompressorInputStream(in)) {
//				final byte[] buffer = new byte[1024];
//				int offset;
//				while (-1 != (offset = zin.read(buffer))) {
//					out.write(buffer, 0, offset);
//				}
//				return out.toString();
//			} catch (final IOException e) {
//				throw new RuntimeException(e);
//			}
//		}


	}
}
